#pragma once
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <memory>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <thread>
#include <fmt/core.h>
#include <functional>
#include "../logger.hpp"

namespace shared {

namespace infra {


class MQClient
{
    using MessageCallback = std::function<void(const char*, size_t)>;
public:
    MQClient(const std::string& user, const std::string& passwd, const std::string& host)
    {
        loop_ = EV_DEFAULT;
        handler_ = std::make_unique<AMQP::LibEvHandler>(loop_);
        std::string url = fmt::format("amqp://{}:{}@{}/", user, passwd, host);
        AMQP::Address address(url);
        connection_ = std::make_unique<AMQP::TcpConnection>(handler_.get(), address);
        channel_ = std::make_unique<AMQP::TcpChannel>(connection_.get());

        ev_async_init(&asyncWatcher_, watcherCallback);  // 初始化异步事件结构asyncWatcher
        ev_async_start(loop_, &asyncWatcher_);           // 将asyncWatcher注册到事件循环中
        loopThread_ = std::thread([this](){
            ev_run(loop_, 0);
        });
    }
    ~MQClient()
    {
        ev_async_send(loop_, &asyncWatcher_);            // 触发异步事件, 回调停止事件循环
        loopThread_.join();
        ev_async_stop(loop_, &asyncWatcher_);            // 清理异步事件结构(管道/fd ...)
        loop_ = nullptr;  // 通过EV_DEFAULT获取时不需要手动释放, ev_loop是libev库内部管理的单例
    }

    void declareComponents(const std::string& exchange, const std::string& queue, 
                           const std::string& routingKey = "routing_key", 
                           AMQP::ExchangeType exchangeType = AMQP::ExchangeType::direct)
    {
        channel_->declareExchange(exchange, exchangeType)
            .onError([](const char* message) {
                LOG_ERROR("声明交换机失败: {}", message);
                exit(1);
            })
            .onSuccess([exchange](){
                LOG_DEBUG("{} 交换机创建成功！", exchange);
            });

        channel_->declareQueue(queue)
            .onError([](const char* message) {
                LOG_ERROR("声明队列失败: {}", message);
                exit(1);
            })
            .onSuccess([queue](){
                LOG_DEBUG("{} 队列创建成功！", queue);
            });

        channel_->bindQueue(exchange, queue, routingKey)
            .onError([exchange, queue](const char *message) {
                LOG_ERROR("{}-{} 绑定失败: ", exchange, queue);
                exit(1);
            })
            .onSuccess([exchange, queue, routingKey](){
                LOG_DEBUG("{}-{}-{} 绑定成功！", exchange, queue, routingKey);
            });
    }

    bool publish(const std::string& exchange, const std::string& msg, 
                const std::string& routingKey = "routing_key")
    {
        LOG_DEBUG("向交换机 {}-{} 发布消息！", exchange, routingKey);
        if(channel_->publish(exchange, routingKey, msg) == false) {
            LOG_ERROR("{} 发布消息失败：", exchange);
            return false;
        }
        return true;
    }

    void consume(const std::string& queue, const MessageCallback& cb)
    {
        LOG_DEBUG("开始订阅 {} 队列消息！", queue);
        channel_->consume(queue, "consume-tag")
            .onReceived([this, cb](const AMQP::Message& message, uint64_t deliveryTag, bool redelivered) {
                // 这里多封一层, 是为了让 MessageCallback 专注与数据的处理, 对于消息本身的其它操作由接收回调本身来做
                cb(message.body(), message.bodySize());
                channel_->ack(deliveryTag);
            })   
            .onError([queue](const char* message) {
                LOG_ERROR("订阅 {} 队列消息失败: {}", queue, message);
                exit(1);
            }); 
    }

private:
    static void watcherCallback(struct ev_loop* loop, ev_async* watcher, int32_t revents) {
        ev_break(loop, EVBREAK_ALL);
    }

private:
    struct ev_loop* loop_;
    struct ev_async asyncWatcher_;
    std::unique_ptr<AMQP::LibEvHandler> handler_;
    std::unique_ptr<AMQP::TcpConnection> connection_;
    std::unique_ptr<AMQP::TcpChannel> channel_;
    std::thread loopThread_;
};

using MQClientPtr = std::shared_ptr<MQClient>;

}}